package com.bayi.mqtt;

import android.content.Context;
import android.os.Handler;
import android.os.Looper;
import android.text.TextUtils;
import android.util.Log;

import com.bayi.mqtt.listener.MessageListener;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * @author Administrator on 2022/12/12.
 * <p>
 * 站在顶峰 看世界
 * 落到谷底 思人生
 */
public class MqttSingle {

    private static final String TAG = MqttSingle.class.getSimpleName();
    private MqttAndroidClient mqttClient;
    private volatile static MqttSingle instance = null;
    private final long connectTime = 30;

    private Context context;
    private String host = "";     // TCP协议
    private String userName = ""; //mqtt用户名称
    private String mqttId = "";//mqtt id
    private String[] mqtt_sub_topic = new String[]{};//mqtt订阅的主题的标识
    private int[] qos = {};//mqtt订阅的主题的标识
    private String[] mqtt_sub_topic_event = new String[]{};//mqtt订阅的主题的标识
    private int[] qos_event = {};//mqtt订阅的主题的标识

    private MessageListener messageListener;

    public MessageListener getMessageListener() {
        return messageListener;
    }

    public void setMessageListener(MessageListener messageListener) {
        this.messageListener = messageListener;
    }

    public static MqttSingle getInstance() {
        if (instance == null) {
            synchronized (MqttSingle.class) {
                if (instance == null) {
                    instance = new MqttSingle();
                }
            }
        }
        return instance;
    }

    public void init(Context context,
                     String host,
                     String userName) {
        this.context = context;
        this.host = host;
        this.userName = userName;
        this.mqttId = UUID.randomUUID().toString().replaceAll("-", "_").substring(0, 13);
        mqttClient = new MqttAndroidClient(context, host, mqttId);
        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                Log.d(TAG, "Connection lost " + cause.toString());
                reconnection(connectTime);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                Log.d(TAG, "Connection arrived " + message.toString());
                if (messageListener != null) {
                    messageListener.messageArrived(topic, message.toString());
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                Log.d(TAG, "deliveryComplete " + token.isComplete());
            }
        });

    }


    public void connect() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(userName);
        try {
            mqttClient.connect(options, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.d(TAG, "Connection success");
                    if (mqtt_sub_topic.length != 0) {
                        subscribe(mqtt_sub_topic, qos);
                    }
                    if (mqtt_sub_topic_event.length != 0) {
                        subscribeEvent(mqtt_sub_topic_event, qos_event);
                    }
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.d(TAG, "Connection failure" + asyncActionToken.getException().getMessage());
                    exception.printStackTrace();
                    reconnection(connectTime);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 重连
     */
    public void reconnection(long delayMillis) {
        handler.removeCallbacks(runnable);
        handler.postDelayed(runnable, delayMillis * 1000);
    }

    private final Handler handler = new Handler(Looper.getMainLooper());
    private final Runnable runnable = this::connect;


    private final Handler handlerSubscribe = new Handler(Looper.getMainLooper());
    private final Runnable runnableSubscribe = () -> {
        subscribe(mqtt_sub_topic, qos);
        subscribeEvent(mqtt_sub_topic_event, qos_event);
    };

    /**
     * 订阅 topic
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String[] topic, int[] qos) {
        this.mqtt_sub_topic = topic;
        this.qos = qos;
        try {
            handlerSubscribe.removeCallbacks(runnableSubscribe);
            if (mqttClient.isConnected()) {
                mqttClient.subscribe(topic, qos, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        Log.d(TAG, "Subscribed to " + topic);
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.d(TAG, "Failed to subscribe " + topic);
                    }
                });
            } else {
                handlerSubscribe.postDelayed(runnableSubscribe, 5 * 1000);
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅 topic
     *
     * @param topic
     * @param qos
     */
    public void subscribeEvent(String[] topic, int[] qos) {
        this.mqtt_sub_topic_event = topic;
        this.qos_event = qos;
        try {
            if (mqttClient.isConnected()) {
                mqttClient.subscribe(topic, qos, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        Log.d(TAG, "Subscribed to " + topic);
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.d(TAG, "Failed to subscribe " + topic);
                    }
                });
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 取消订阅
     */
    public void unsubscribe() {
        try {
            if (mqtt_sub_topic.length != 0) {
                mqttClient.unsubscribe(mqtt_sub_topic, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        Log.d(TAG, "Unsubscribed to " + mqtt_sub_topic);
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.d(TAG, "Failed to unsubscribe " + mqtt_sub_topic);
                    }
                });
            }
            if (mqtt_sub_topic_event.length != 0) {
                mqttClient.unsubscribe(mqtt_sub_topic_event, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        Log.d(TAG, "Unsubscribed to " + mqtt_sub_topic_event);
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.d(TAG, "Failed to unsubscribe " + mqtt_sub_topic_event);
                    }
                });
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    /**
     * 发布消息
     *
     * @param topic
     * @param msg
     */
    public void publish(String topic, String msg) {
        if (TextUtils.isEmpty(topic) || TextUtils.isEmpty(msg)) {
            return;
        }
        new Handler(Looper.getMainLooper()).post(() -> {
            try {
                MqttMessage message = new MqttMessage();
                message.setPayload(msg.getBytes(StandardCharsets.UTF_8));
                message.setQos(1);
                message.setRetained(false);
                mqttClient.publish(topic, message, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        Log.d(TAG, String.format("%s published to %s", msg, topic));
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.d(TAG, String.format("Failed to publish $msg to %s", topic));
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 断开 MQTT 连接
     */
    public void disconnect() {
        try {
            mqttClient.disconnect(null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.d(TAG, "Disconnected");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.d(TAG, "Failed to disconnect");
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
